package cn.ac.iie.di.ban.ycl.cleaner.handler;

import cn.ac.iie.di.ban.ycl.cleaner.CleanerMain;
import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import org.ahocorasick.trie.Emit;
import org.ahocorasick.trie.Trie;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.http.client.fluent.Request;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zmc <zhangmingcheng@iie.ac.cn>
 * @date 2016-9-8 13:53:12
 * @version V2.0
 * @desc 数据去重
 */
public class CleanerHandler implements Runnable {

    final Logger LOG = LoggerFactory.getLogger(CleanerHandler.class);
    BlockingQueue<REMessage> queue;
    private String[] columnNames;
    private REFieldType[] types;
    private String docSchemaName;
    private Map<String, String> userDesc;
    private List<Object[]> datas;
    private final List<Object> distinctFields;
    private final boolean isNeedDeduplication;
    private final List<Object> filterKeyWords;
    private final int filterContentSize;
    private final boolean isNeedFilter;
    private final String[] urls;
    private List<Object[]> rubbishDatas;
    private final int batchNum;
    private final int timeOut;

    public CleanerHandler(Configuration configuration, BlockingQueue queue) {
        this.queue = queue;
        this.isNeedDeduplication = configuration.getBoolean(Constants.DEDUPLICATION_NECESSARY);
        this.distinctFields = configuration.getList(Constants.DEDUPLICATION_FIELDS);
        this.isNeedFilter = configuration.getBoolean(Constants.FILTER_NECESSARY);
        this.filterKeyWords = configuration.getList(Constants.FILTER_CONTENT_KEYWORDS);//关键词列表
        this.filterContentSize = configuration.getInt(Constants.FILTER_CONTENT_SIZE);//内容长度
        this.urls = configuration.getStringArray(Constants.DEDUPLICATION_URL);
        this.batchNum = configuration.getInt(Constants.HANDLER_BATCH_NUM);
        this.timeOut = configuration.getInt(Constants.HANDLER_TIME_OUT);
    }

    @Override
    public void run() {
        while (true) {
            try {
                REMessage message = queue.poll(3, TimeUnit.SECONDS);
                if (message == null) {
                    LOG.warn("there is not message from the receiver queue!");
                    continue;
                }
                columnNames = message.colNames;
                types = message.types;
                docSchemaName = message.docSchemaName;
                userDesc = message.userDesc;
                datas = message.data;
                LOG.info("receive messages num: {}", message.data.size());
                handle();
            } catch (Exception ex) {
                LOG.error("clean data has error! {}", ExceptionUtils.getFullStackTrace(ex));
            }
        }
    }

    private void handle() throws InterruptedException, Exception {
        if (isNeedDeduplication) { //数据去重
            String deduplicationValues = getDeduplicationValues();
            if (deduplicationValues.isEmpty()) {
                LOG.error("deduplication values is empty!");
                return;
            } else {
                LOG.debug("deduplication values: \n{}", deduplicationValues);
            }
            String response = deduplicationService(deduplicationValues);
            LOG.debug("deduplication result:\n{}", response);
            String[] split = response.split("\n");
            List<String> flags = Arrays.asList(split);
            deleteDuplicationDatas(flags);
            REMessage deduplicationMessage = new REMessage(columnNames, types, datas, docSchemaName, userDesc);
            CleanerMain.deduplicationQueue.put(deduplicationMessage);
        }

        if (isNeedFilter) {//数据过滤
            rubbishDatas = new ArrayList<>();
            filter();
            REMessage filterMessage = new REMessage(columnNames, types, datas, docSchemaName, userDesc);
            CleanerMain.filterQueue.put(filterMessage);
            REMessage rubbishMessage = new REMessage(columnNames, types, rubbishDatas, docSchemaName, userDesc);
            CleanerMain.rubbishQueue.put(rubbishMessage);
        }
    }

    //获取需要去重的字段值
    public String getDeduplicationValues() {
        StringBuilder sb = new StringBuilder();
        List<String> columnList = Arrays.asList(columnNames);
        for (Object[] data : this.datas) {
            boolean isFirst = true;
            for (Iterator<Object> it = distinctFields.iterator(); it.hasNext();) {
                String distinctField = (String) it.next();
                int index = columnList.indexOf(distinctField);
                if (index < 0) {
                    LOG.info(distinctField);
                    LOG.error("the distinct field: {} does not exists!", distinctField);
                }
                Object distinctFieldValue = data[index];
                if (distinctFieldValue == null || distinctFieldValue.toString().isEmpty()) {
                    LOG.info("columnames:{}", Arrays.toString(columnNames));
                    LOG.info("data:{}", Arrays.toString(data));
                    LOG.error("the value of distinct field: {} is null or empty!", distinctField);
                }
                if (!isFirst) {
                    sb.append("^");
                }
                sb.append(distinctFieldValue);
                isFirst = false;
            }
            sb.append("\n");
        }
        return sb.toString();
    }

    public void deleteDuplicationDatas(List<String> flags) {
        List<Object[]> newMessage = new ArrayList<>();
        for (int i = 0; i < flags.size(); i++) {
            if (flags.get(i).equals("0")) {
                newMessage.add(datas.get(i));
            }
        }
        this.datas = newMessage;
    }

    private String deduplicationService(String content) {
        String response = null;
        String url;
        int index;
        while (true) {
            index = new Random().nextInt(urls.length);
            url = urls[index];
            try {
                response = Request.Post(url)
                        .bodyByteArray(content.getBytes())
                        .connectTimeout(5000)
                        .socketTimeout(3000)
                        .execute()
                        .returnContent().asString();
            } catch (IOException ex) {
                LOG.warn("request url {} error! \n"
                        + ExceptionUtils.getFullStackTrace(ex), url);
            }
            if (response == null || response.trim().isEmpty()) {
                LOG.error("response content is null，please check the url:{}", url);
            } else if (!response.contains("1") && !response.contains("0")) {
                LOG.error("response content format is error，please check the url:{},content is", url, response);
            } else {
                break;
            }
            try {
                Thread.sleep(5000);
            } catch (InterruptedException ex) {
            }
        }
        return response;
    }

    /**
     * 过滤规则：cont字段对应的文本内容长度如果小于等于50，就与配置的关键词列表匹配，命中则为有效数据； 反之长度大于50，则为命中数据；
     * 包含图片数据的则为命中数据
     *
     * @throws java.lang.Exception
     */
    public void filter() throws Exception {
        List<Object[]> newMessage = new ArrayList<>();
        List<String> columnList = Arrays.asList(columnNames);
        int wxauIndex = columnList.indexOf("wx_au");
        if (wxauIndex < 0) {
            throw new Exception("wx_au field does not exists! " + ArrayUtils.toString(columnList));
        }
        int contIndex = columnList.indexOf("cont");
        if (contIndex < 0) {
            throw new Exception("cont field does not exists! " + ArrayUtils.toString(columnList));
        }
        int wxSfromIndex = columnList.indexOf("wx_sfrom");
        if (wxSfromIndex < 0) {
            throw new Exception("wx_sfrom field does not exists! " + ArrayUtils.toString(columnList));
        }

        Trie trie = new Trie();
        for (Iterator<Object> it = filterKeyWords.iterator(); it.hasNext();) {
            String keyword = (String) it.next();
            String keywordUnicode = chToUnicode(keyword);
            trie.addKeyword(keywordUnicode);
        }
        for (Object[] data : this.datas) {
            //推送消息无ID
            String wxSfrom = (String) data[wxSfromIndex];
            if (wxSfrom == null || wxSfrom.isEmpty()) {
//                LOG.warn("the value of cont field  is null or empty!");
                rubbishDatas.add(data);
                continue;
            }
            //包含图片
            String[] wxauContent = (String[]) data[wxauIndex];
            if (wxauContent.length == 1 && wxauContent[0].isEmpty()) {
//                LOG.warn("the value of wxau field  is null or empty!");
            } else {
                LOG.debug("wx_au:" + Arrays.toString(wxauContent));
                newMessage.add(data);
                continue;
            }

            String content = (String) data[contIndex];
            if (content == null || content.isEmpty()) {
//                LOG.warn("the value of cont field  is null or empty!");
                rubbishDatas.add(data);
            } else if (content.length() <= this.filterContentSize) { //字符串长度《=50,匹配关键词
                Collection<Emit> emits = trie.parseText(chToUnicode(content));
                if (!emits.isEmpty() && emits.size() > 0) {
                    newMessage.add(data);
                } else {
                    rubbishDatas.add(data);
                }
            } else {//字符串长度>50
                newMessage.add(data);
            }
        }
        this.datas = newMessage;
    }

    private String chToUnicode(String content) {
        char[] utfBytes = content.toCharArray();
        String unicodeBytes = "";
        for (int byteIndex = 0; byteIndex < utfBytes.length; byteIndex++) {
            String hexB = Integer.toHexString(utfBytes[byteIndex]);
            if (hexB.length() <= 2) {
                hexB = "00" + hexB;
            }
            unicodeBytes = unicodeBytes + "\\u" + hexB;
        }
        return unicodeBytes;
    }

}
